package com.breeze.components.message.delay.holder;

import com.alibaba.fastjson2.JSON;
import com.breeze.components.message.delay.msg.DelayMessage;
import jakarta.annotation.Resource;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.stereotype.Component;

/**
 * Redisson 实现
 *
 * @author breeze
 */
@Slf4j
@Component
@ConditionalOnClass(RedissonClient.class)
@ConditionalOnMissingBean(DelayQueueHolder.class)
public class RedisDelayQueueHolder implements DelayQueueHolder {

  /** 队列缓存 */
  private final Map<String, RDelayedQueue<String>> delayedQueueMap = new ConcurrentHashMap<>(16);

  @Resource private RedissonClient redissonClient;

  @Override
  public void addJob(Object value, long delay, TimeUnit timeUnit, String queueName) {
    try {
      log.info(
          "添加到延时队列 queue:{}, value:{}, delay:[{}-{}]", queueName, value, delay, timeUnit.name());
      RDelayedQueue<String> delayedQueue = this.initDelayQueue(queueName);
      delayedQueue.offer(JSON.toJSONString(value), delay, timeUnit);
    } catch (Exception e) {
      log.error("添加到延时队列失败:value:{} queueName:{}", value, queueName, e);
    }
  }

  @Override
  public void addJob(Object value, LocalDateTime delayTime, String queueName) {
    long millis = Duration.between(LocalDateTime.now(), delayTime).toMillis();
    if (millis <= 0) {
      log.error("添加到延时队列失败,延迟时间<=0:value:{} queueName:{}", value, queueName);
      return;
    }
    addJob(value, millis, TimeUnit.MILLISECONDS, queueName);
  }

  @Override
  public <T extends DelayMessage> List<T> removeJob(String queueName, T t, Class<T> classType) {
    RDelayedQueue<String> delayedQueue = this.initDelayQueue(queueName);
    log.info("移除延时队列，queueName:{}, taskId:{}", queueName, t.taskIdentity());
    List<String> dataList =
        delayedQueue.stream()
            .filter(
                data -> {
                  T k = JSON.parseObject(data, classType);
                  if (k == null) {
                    return false;
                  }
                  if (k.taskIdentity().equals(t.taskIdentity())) {
                    log.info("匹配到需要删除的队列数据:{}", JSON.toJSONString(data));
                    return true;
                  }
                  return false;
                })
            .toList();
    log.info("任务ID:{}匹配到删除的记录数:{}", t.taskIdentity(), dataList.size());
    delayedQueue.removeAll(dataList);
    return dataList.stream()
        .map(data -> JSON.parseObject(data, classType))
        .collect(Collectors.toList());
  }

  @Override
  public void destroyQueue(String queueName) {
    RDelayedQueue<String> delayedQueue = delayedQueueMap.get(queueName);
    if (delayedQueue != null) {
      delayedQueue.destroy();
    }
  }

  @Override
  public BlockingDeque<String> getQueue(String queueName) {
    // 重启后初始化
    this.initDelayQueue(queueName);
    return redissonClient.getBlockingDeque(queueName);
  }

  @Override
  public String take(String queueName) throws Exception {
    return getQueue(queueName).take();
  }

  /**
   * 延迟队列初始化
   *
   * @param queueName 队列名称
   * @return 延迟队列
   */
  @Override
  public RDelayedQueue<String> initDelayQueue(String queueName) {
    RDelayedQueue<String> delayedQueue = delayedQueueMap.get(queueName);
    if (null == delayedQueue) {
      RBlockingDeque<String> blockingDeque = redissonClient.getBlockingDeque(queueName);
      delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
      delayedQueueMap.put(queueName, delayedQueue);
    }
    return delayedQueue;
  }
}
